package com.niit.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object Spark_stream_UpdateStateByKey {

  def main(args: Array[String]): Unit = {
    //初始化环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("sparkStream")
    //初始化SparkStreamingContext            采集周期3秒
    val ssc = new StreamingContext(sparkConf,Seconds(3))
    ssc.sparkContext.setLogLevel("ERROR")
    ssc.checkpoint("BD1")//将统计的状态进行储存，该存储要落盘（写入在磁盘）
    //通过监控端口创建DStream ,读取数据为一行一行  监听本机 9999端口发送的数据
    var lines:ReceiverInputDStream[String] =  ssc.socketTextStream("localhost",9999)

    //将单词映射（转换）元组
    var wordOne =  lines.map( (_,1) )

    /*
    updateStateByKey:根据Key对数据的状态进行更新
    传递的参数中含有两个值
    第一个值，表示相同key的value数据
    第二个值，表示缓冲区相同key的value数据  保存上次的统计结果
     */
    val state =  wordOne.updateStateByKey(
      (seq:Seq[Int],buff:Option[Int])=>{// buff.getOrElse(0)  如果里面有值则得到相应的值，如果没有则返回 0
        val newCount = buff.getOrElse(0) + seq.sum  //缓冲区里面的value + 本次统计序列的和
        Option(newCount)
      }
    )

    //打印
    state.print()

    //1.开启采集器
    ssc.start()
    //2.等待采集器关闭
    ssc.awaitTermination()


  }

}
